package com.giraffe.mqtt.queue;


import com.giraffe.mqtt.bean.SendMqttMessage;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class MessageHandler {

    @EventListener
    @Async("asyncMessageThread")
    public void onEvent(MessageEvent messageEvent) throws Exception {
        SendMqttMessage message = messageEvent.getMessage();
        if (message.getChannel().isActive()) {
            switch (message.getConfirmStatus()) {
                case PUB:
                    pubMessage(message.getChannel(), message);
                    break;
                case PUBREL:
                    sendAck(MqttMessageType.PUBREL, message);
                    break;
                case PUBREC:
                    sendAck(MqttMessageType.PUBREC, message);
                    break;
            }
        }
    }

    private void pubMessage(Channel channel, SendMqttMessage mqttMessage) {
        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, true, mqttMessage.getQos(), mqttMessage.isRetain(), 0);
        MqttPublishVariableHeader mqttPublishVariableHeader = new MqttPublishVariableHeader(mqttMessage.getTopic(), mqttMessage.getMessageId());
        MqttPublishMessage mqttPublishMessage = new MqttPublishMessage(mqttFixedHeader, mqttPublishVariableHeader, Unpooled.wrappedBuffer(mqttMessage.getByteBuf()));
        channel.writeAndFlush(mqttPublishMessage);
    }

    protected void sendAck(MqttMessageType type, SendMqttMessage mqttMessage) {
        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(type, true, MqttQoS.AT_LEAST_ONCE, false, 0x02);
        MqttMessageIdVariableHeader from = MqttMessageIdVariableHeader.from(mqttMessage.getMessageId());
        MqttPubAckMessage mqttPubAckMessage = new MqttPubAckMessage(mqttFixedHeader, from);
        mqttMessage.getChannel().writeAndFlush(mqttPubAckMessage);
    }

}
